package cn.ljy.kafkastudy.producer;

import cn.ljy.kafkastudy.Consts;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.util.JSONPObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;

/**
 * @description:
 * @author: James
 * @create: 2022-03-05 13:51
 */
@Slf4j
@RestController
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @RequestMapping("/send")
    public void send() {
        //发送功能就一行代码~
        ListenableFuture<SendResult<String,String>> future=kafkaTemplate.send(Consts.TOPIC_NAME,  "key", "test message send~");
        try {
            ObjectMapper om=new ObjectMapper();
//            只要发送给broker即可返回
            SendResult<String,String> result=future.get();
            log.info("producerRecord:"+result.getProducerRecord());
            RecordMetadata metadata=result.getRecordMetadata();

            log.info("recordMetadata:");
            log.info("hasOffset:"+metadata.hasOffset()+",offset:"+metadata.offset()
                    +",topic:"+metadata.topic()+",partition:"+metadata.partition()
                    +",hasTimestamp:"+metadata.hasTimestamp()+",timestamp:"+metadata.timestamp());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

    @RequestMapping("/send2")
    public void send2() {
        //发送功能就一行代码~
        ListenableFuture<SendResult<String,String>> future=kafkaTemplate.send(Consts.TOPIC_NAME2,  "key", "test message send~");

    }
}
